package com.atguigu.ad.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author yhm
 * @create 2023-02-13 14:33
 */
public class TimestampInterceptor implements Interceptor {
    private Pattern pattern;

    @Override
    public void initialize() {
        pattern =  Pattern.compile(".*t=(\\d{13}).*");
    }

    @Override
    public Event intercept(Event event) {
        // 1.对数据进行简单的清洗过滤
        // 2.将数据中的path中的t参数时间戳写入到header信息中
        //"2023-01-07 23:54:37""GET""/ad/juliang/click?id=12&t=1673106876041&ip=103.57.58.212&ua=Mozilla/5.0%20(X11;%20U;%20Gentoo%20Linux%20x86_64;%20pl-PL;%20rv:1.8.1.7)%20Gecko/20070914%20Firefox/2.0.0.7""200""103.125.199.121"
        try {
            String log = new String(event.getBody(), StandardCharsets.UTF_8);
            String substring = log.substring(1, log.length() - 1);
            String newLog = substring.replaceAll("\"\u0001\"", "\u0001");
            Matcher matcher = pattern.matcher(log);
            if (matcher.matches()){
                String ts = matcher.group(1);
                Map<String, String> headers = event.getHeaders();
                headers.put("timestamp",ts);
            }else {
                return null;
            }
            event.setBody(newLog.getBytes());
        }catch (Exception e){
            e.printStackTrace();
            return null;
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        Iterator<Event> iterator = events.iterator();
        while (iterator.hasNext()){
            Event next = iterator.next();
            if (intercept(next) == null) {
                iterator.remove();
            }
        }
        return events;
    }

    @Override
    public void close() {

    }


    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new TimestampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
